package com.dd.seckill.cluster.service.internal.impl;

import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.dd.seckill.cluster.config.cache.caffeine.SeckillGoodsCache;
import com.dd.seckill.cluster.config.exception.BusinessException;
import com.dd.seckill.cluster.config.exception.RedisScriptException;
import com.dd.seckill.cluster.config.rabbitmq.MQSeckillMsgSender;
import com.dd.seckill.cluster.config.redis.lua.RedisScriptPreLoader;
import com.dd.seckill.cluster.mapper.SeckillGoodsMapper;
import com.dd.seckill.cluster.mapper.SeckillOrderMapper;
import com.dd.seckill.cluster.pojo.DO.SeckillGoods;
import com.dd.seckill.cluster.pojo.DO.SeckillOrder;
import com.dd.seckill.cluster.pojo.annotation.RateLimit;
import com.dd.seckill.cluster.pojo.dto.SeckillMessage;
import com.dd.seckill.cluster.pojo.req.SeckillReq;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.List;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author liuxianmeng
 * @CreateTime 2025/9/6 23:51
 * @Description 秒杀服务
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class SeckillServiceImpl {

    private final SeckillGoodsMapper seckillGoodsMapper;
    private final SeckillOrderMapper seckillOrderMapper;

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private RedisScriptPreLoader redisScriptPreLoader;

    @Resource
    MQSeckillMsgSender mqSeckillMsgSender;

    // Lua脚本执行结果常量
    private static final int RESULT_SUCCESS = 1;
    private static final int RESULT_STOCK_OUT = -1;
    private static final int RESULT_USER_EXISTS = -2;
    private static final int RESULT_SYSTEM_ERROR = -3;

    private static AtomicInteger USER_EXISTS_COUNTER = new AtomicInteger(0);
    private static AtomicInteger SYSTEM_ERROR_COUNTER = new AtomicInteger(0);

    /**
     * 秒杀服务
     *
     * @param req SeckillReq
     */
    //@RateLimit(
    //        key = "seckill",
    //        limit = 5,        // 时间窗口内最大请求数
    //        window = 60,         // 时间窗口（秒）
    //        message = "秒杀请求过于频繁，请稍后再试"
    //)
    public void seckillV2(SeckillReq req) {
        long goodsId = req.getGoodsId();
        long userId = req.getUserId();

        // 使用hash tag确保key在同一slot
        String goodsKey = "seckill:goods:{" + goodsId + "}";
        String stockKey = "seckill:stock:{" + goodsId + "}";
        String userSetKey = "seckill:success:users:{" + goodsId + "}";

        try {
            // 执行Lua脚本
            List<Object> result = executeSeckillScript(goodsId, userId);

            // 解析Lua脚本执行结果
            int code = Integer.parseInt(result.get(0).toString());
            String message = result.get(1).toString();
            String stockData = result.get(2).toString();

            switch (code) {
                case RESULT_SUCCESS:
                    // Lua脚本执行成功，继续数据库操作
                    int remainingStock = Integer.parseInt(stockData);
                    SeckillMessage seckillMessage = SeckillMessage.builder()
                            .goodsId(goodsId)
                            .userId(userId)
                            .stockKey(stockKey)
                            .goodsKey(goodsKey)
                            .userSetKey(userSetKey)
                            .build();
                    //mqSeckillMsgSender.senderMessage(JSONUtil.toJsonStr(seckillMessage));
                    break;
                case RESULT_STOCK_OUT:
                    throw new BusinessException(500001, "秒杀失败，商品已售罄。");
                case RESULT_USER_EXISTS:
                    log.warn("USER_EXISTS_COUNTER = {}", USER_EXISTS_COUNTER.incrementAndGet());
                    throw new BusinessException(500002, "秒杀失败，您已购买过此商品。");
                case RESULT_SYSTEM_ERROR:
                    log.warn("RESULT_SYSTEM_ERROR = {}", USER_EXISTS_COUNTER.incrementAndGet());
                    SYSTEM_ERROR_COUNTER.incrementAndGet();
                default:
                    log.warn("RESULT_SYSTEM_ERROR = {}", USER_EXISTS_COUNTER.incrementAndGet());
                    SYSTEM_ERROR_COUNTER.incrementAndGet();
                    throw new BusinessException(500, "秒杀失败，系统繁忙。");
            }
        } catch (RedisScriptException e) {
            log.error("Redis script execution failed for goodsId: {}, userId: {}", goodsId, userId, e);
            throw new BusinessException(500, "秒杀失败，系统繁忙");
        }
    }

    /**
     * 执行秒杀Lua脚本 - 集群版
     */
    private List<Object> executeSeckillScript(long goodsId, long userId) {
        try {
            // 使用hash tag确保key在同一slot
            String stockKey = "seckill:stock:{" + goodsId + "}";

            // 优先使用SHA1执行脚本
            return stringRedisTemplate.execute(
                    new DefaultRedisScript<List<Object>>() {
                        @Override
                        public String getSha1() {
                            return redisScriptPreLoader.getSeckillScriptSha1();
                        }
                        @Override
                        public Class<List<Object>> getResultType() {
                            return (Class<List<Object>>) (Class<?>) List.class;
                        }
                    },
                    Collections.singletonList(stockKey), // 集群模式下需要指定key定位slot
                    String.valueOf(goodsId), String.valueOf(userId)
            );
        } catch (Exception e) {
            // SHA1执行失败，回退到原始脚本执行
            log.warn("Execute script by SHA1 failed, fallback to original script", e);
            try {
                DefaultRedisScript<List<Object>> script = new DefaultRedisScript<>();
                script.setScriptText(getSeckillScriptContent());
                script.setResultType((Class<List<Object>>) (Class<?>) List.class);

                String stockKey = "seckill:stock:{" + goodsId + "}";

                return stringRedisTemplate.execute(script,
                        Collections.singletonList(stockKey),
                        String.valueOf(goodsId), String.valueOf(userId));
            } catch (Exception ex) {
                throw new RedisScriptException("Failed to execute seckill script", ex);
            }
        }
    }

    /**
     * 秒杀服务 数据库持久化
     *
     * [事务传播] REQUIRED 内部"子事务"要么全部成功 要么全部失败
     *
     * @param seckillMessage SeckillMessage
     */
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public void processDatabaseOperations(SeckillMessage seckillMessage) {
        // 1 库存校验（即便Redis已经做过校验）
        SeckillGoods seckillGoods = seckillGoodsMapper.queryWithPessmisticLock(seckillMessage.getGoodsId());
        if(seckillGoods.getAvailableStock() < 1) {
            // 数据库不足，回滚Redis操作
            rollbackRedisOperations(seckillMessage.getStockKey(), seckillMessage.getUserSetKey(), seckillMessage.getUserId());
            throw new BusinessException(500, "秒杀失败，库存不足");
        }

        LambdaQueryWrapper<SeckillOrder> lqw = new LambdaQueryWrapper<>();
        lqw.eq(SeckillOrder::getOrderNo, seckillMessage.getUserId() + "-" +seckillMessage.getGoodsId());
        List<SeckillOrder> seckillOrders = seckillOrderMapper.selectList(lqw);
        if(!seckillOrders.isEmpty()) {
            System.out.println("该订单号已存在：" + seckillMessage.getUserId() + "-" +seckillMessage.getGoodsId());
            return;
        }

        // 2 数据库层面再次校验减库存
        int decrRst = seckillGoodsMapper.decrStock(seckillMessage.getGoodsId());
        if (decrRst != 1) {
            // 数据库减库存失败，回滚Redis操作
            rollbackRedisOperations(seckillMessage.getStockKey(), seckillMessage.getUserSetKey(), seckillMessage.getUserId());
            throw new BusinessException(500, "秒杀失败，库存不足");
        }

        // 3 生成新订单
        //BigDecimal seckillPrice = getSeckillPriceFromRedis(seckillMessage.getGoodsKey());
        BigDecimal seckillPrice = getSeckillPriceFromCaffeine(seckillMessage.getGoodsId()).getSeckillPrice();
        SeckillOrder order = SeckillOrder.builder()
                .orderNo(generateRandomStr(16))
                .seckillPrice(seckillPrice)
                .status(new Byte(String.valueOf(0)))
                .createTime(LocalDateTime.now())
                .goodsId(seckillMessage.getGoodsId())
                .userId(seckillMessage.getUserId())
                .quantity(1)
                .build();
        seckillOrderMapper.insert(order);
    }

    /**
     * 回滚Redis操作
     */
    private void rollbackRedisOperations(String stockKey, String userSetKey, Long userId) {
        try {
            // 恢复库存
            redisTemplate.opsForValue().increment(stockKey);
            // 移除用户购买记录
            redisTemplate.opsForSet().remove(userSetKey, String.valueOf(userId));
            log.info("Rollback Redis operations for stockKey: {}, userId: {}", stockKey, userId);
        } catch (Exception e) {
            log.error("Failed to rollback Redis operations for stockKey: {}, userId: {}", stockKey, userId, e);
            // 这里可以选择记录日志或发送告警，但不应该阻止主流程
        }
    }

    private SeckillGoods getSeckillPriceFromCaffeine(Long goodsId) {
        return SeckillGoodsCache.seckillGoodsCache.get(goodsId, seckillGoodsMapper::selectById);
    }

    /**
     * 获取Lua脚本内容
     */
    private String getSeckillScriptContent() {
        return "local goodsId = ARGV[1]\n" +
                "local userId = ARGV[2]\n" +
                "\n" +
                "local stockKey = \"seckill:stock:{\" .. goodsId .. \"}\"\n" +
                "local userSetKey = \"seckill:success:users:{\" .. goodsId .. \"}\"\n" +
                "\n" +
                "local result = {}\n" +
                "\n" +
                "local stock = redis.call('DECR', stockKey)\n" +
                "if stock < 0 then\n" +
                "    redis.call('INCR', stockKey)\n" +
                "    result[1] = tostring(-1)\n" +
                "    result[2] = \"库存不足\"\n" +
                "    result[3] = tostring(stock)\n" +
                "    return result\n" +
                "end\n" +
                "\n" +
                "local isMember = redis.call('SISMEMBER', userSetKey, userId)\n" +
                "if isMember == 1 then\n" +
                "    redis.call('INCR', stockKey)\n" +
                "    result[1] = tostring(-2)\n" +
                "    result[2] = \"用户已购买\"\n" +
                "    result[3] = tostring(stock)\n" +
                "    return result\n" +
                "end\n" +
                "\n" +
                "local addResult = redis.call('SADD', userSetKey, userId)\n" +
                "if addResult == 0 then\n" +
                "    redis.call('INCR', stockKey)\n" +
                "    result[1] = tostring(-3)\n" +
                "    result[2] = \"系统异常\"\n" +
                "    result[3] = tostring(stock)\n" +
                "    return result\n" +
                "end\n" +
                "\n" +
                "result[1] = tostring(1)\n" +
                "result[2] = \"成功\"\n" +
                "result[3] = tostring(stock)\n" +
                "return result";
    }

    /**
     * 生成随机字符串
     *
     * @param length 字符串的长度
     * @return 生成随机的字符串
     */
    private static String generateRandomStr(Integer length) {
        Random random = new Random();
        StringBuilder tokenBuilder = new StringBuilder(length);
        String CHAR_SET = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789";
        for (int i = 0; i < length; i++) {
            int index = random.nextInt(CHAR_SET.length());
            tokenBuilder.append(CHAR_SET.charAt(index));
        }
        return tokenBuilder.toString();
    }
}
